package com.sunday.common.mq.rabbit.study.base.e14_ConsumerCancelNotification;

import com.rabbitmq.client.*;
import com.sunday.common.mq.rabbit.study.base.ConnectionUtil;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;

/**
 * https://www.rabbitmq.com/consumer-cancel.html
 * 消费者取消通知
 * 当通道从队列中使用时，有各种 可能导致消费停止的原因。其中之一 这些显然是如果客户端在同一通道上发出 basic.cancel 时，
 * 这将 导致消费者被取消并且服务器回复 与基本取消-确定。其他事件，例如 正在删除的队列，或在群集方案中，
 * 队列所在的节点发生故障，将导致 消费将被取消，但客户端渠道不会 了解情况，这通常是无益的。
 * 为了解决这个问题，我们引入了一个扩展，其中 经纪人将向客户发送一个基本取消，如果这种意外的消费者取消。这 在代理从客户端收到 basic.cancel 的情况下不发送。
 * AMQP 0-9-1 默认情况下，客户端不希望从代理接收 basic。cancel 方法 异步，因此为了启用此行为， 客户端必须在 它的客户端属性，其中有一个键consumer_cancel_notify和一个布尔值 true。
 * 有关详细信息，请参阅有关功能的部分。
 * 我们支持的客户端默认将此功能提供给 代理，因此将由代理发送异步 basic.cancel 方法，它们 呈现给使用者回调。例如，在我们的Java中 客户端，消费者接口有一个句柄取消回调，可以覆盖 通过子类化 DefaultConsumer 类：
 * 客户端为已 意外取消（例如，由于队列删除）。由 定义，客户端之间可能存在竞争 发出 basic.cancel，代理发送 输出异步通知。在这种情况下，经纪人 收到 basic.cancel 时不会出错，并像往常一样回复 basic.cancel-ok。
 * <p>
 * 使用者取消和复制队列
 * 支持消费者取消通知的客户端将始终是 当队列被删除或变为 不能利用的。消费者可要求取消 当复制队列的领导者更改时。
 */
@Slf4j
public class ConsumerCancelNotification {

    @Test
    @SneakyThrows
    public void consumerCancelNotification() {

        String QUEUE_NAME = "consumer_cancel_notification_queue";

        ConnectionFactory cf = ConnectionUtil.factory();
        try (Connection connection = cf.newConnection()) {
            Channel channel = connection.createChannel();

            /**
             * 直接划重点 - 仲裁队列属性
             * 有害消息处理
             * 仲裁队列支持处理有害消息， 即，导致使用者重复对传递重新排队的消息（可能是由于使用者故障） 这样消息永远不会被完全使用并得到肯定的确认，以便可以将其标记为 由兔子删除。
             * 仲裁队列跟踪不成功的传递尝试次数，并将其公开在 包含在任何重新传递的邮件中的“X 传递计数”标头。
             * 可以使用策略参数 delivery-limit 为队列设置传递限制。
             * 当消息返回的次数超过限制时，消息将被丢弃或死信（如果配置了 DLX）。
             */
//            Map<String, Object> args = new HashMap<>();
//            args.put("x-delivery-limit", 3);

            channel.queueDelete(QUEUE_NAME);
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);

            channel.basicPublish("", QUEUE_NAME, null, "test".getBytes(StandardCharsets.UTF_8));

            channel.basicConsume(QUEUE_NAME, false, new DefaultConsumer(channel) {

                @Override
                public void handleCancel(String consumerTag) {
                    // consumer has been cancelled unexpectedly
                    log.info("handleCancel consumerTag {}", consumerTag);
                }

                @Override
                public void handleDelivery(String consumerTag,
                                           Envelope envelope,
                                           AMQP.BasicProperties properties,
                                           byte[] body) throws IOException {
                    log.info(" consumerTag : {} ", consumerTag);
                    log.info(" envelope : {} ", envelope);
                    log.info(" properties : {} ", properties);
                    String message = new String(body, "UTF-8");
                    log.info(" [x] Received '{}'", message);
//                    this.getChannel().basicNack(envelope.getDeliveryTag(), false, false);
                    this.getChannel().basicReject(envelope.getDeliveryTag(), false);
                }

                /**
                 * 使用手动确认时，重要的是要考虑确认的线程。如果它与接收传递的线程不同（例如，Consumer＃handleDelivery 委托传递处理到另一个线程），
                 * 则将多个参数设置为true进行确认是不安全的，并且将导致双重确认，因此会导致通道级协议异常关闭频道。一次确认一条消息可能是安全的。
                 */
                @Override
                public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {
                    log.info("handleShutdownSignal consumerTag {}, ShutdownSignalException {}", consumerTag, sig.toString());
                }
            });

            TimeUnit.SECONDS.sleep(2);

        }
    }

}
